package p.ithorns.framework.redis.stream;

import org.springframework.data.domain.Range;
import org.springframework.data.redis.connection.stream.*;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.stereotype.Component;

import java.time.Duration;
import java.util.ArrayList;
import java.util.List;
import java.util.Map;
import java.util.stream.Collectors;
import java.util.stream.Stream;

/**
 * RedisUtil
 * TODO
 *
 * @author ithorns
 * @version 1.0.0
 * @date 2022/11/22 20:05
 */
@Component
public class RedisStreamKit {

    private final RedisTemplate<String, Object> template;

    public RedisStreamKit(RedisTemplate<String, Object> template) {
        this.template = template;
    }

    public RecordId addStream(String key, Map<String, Object> message) {
        // xadd stream
        return template.opsForStream().add(key, message);
    }

    public String addGroup(String key, String groupName) {
        // xgroup create stream group $
        return template.opsForStream().createGroup(key, groupName);
    }

    public Long delField(String key, String fieldId) {
        return template.opsForStream().delete(key, fieldId);
    }

    /**
     * 读取所有stream消息
     *
     * @param key
     * @return
     */
    public List<MapRecord<String, Object, Object>> getAllStream(String key) {
        List<MapRecord<String, Object, Object>> range = template.opsForStream()
                .range(key, Range.open("-", "+"));
        if (range == null) {
            return null;
        }
        for (MapRecord<String, Object, Object> mapRecord : range) {
            template.opsForStream().delete(key, mapRecord.getId());
        }
        return range;
    }

    public List<MapRecord<String, Object, Object>> getStream(String key) {
        return template.opsForStream()
                .read(StreamReadOptions.empty()
                        .block(Duration.ofMillis(1000 * 30)).count(2), StreamOffset.latest(key));
    }

    public List<MapRecord<String, Object, Object>> getStreamByGroup(String key,
                                                                    String groupName,
                                                                    String consumerName) {
        return template.opsForStream()
                .read(
                        Consumer.from(groupName, consumerName),
                        StreamReadOptions.empty(),
                        StreamOffset.create(key, ReadOffset.lastConsumed())
                );
    }

    public boolean hasKey(String key) {
        Boolean aBoolean = template.hasKey(key);
        return aBoolean == null ? false : aBoolean;
    }

    public static List<List<String>> cutApart(int max, List<String> list){
        List<List<String>> resultList = new ArrayList<>();
        Stream.iterate(0, n->n+1).limit(countStep(list.size(), max)).forEach(i->{
            resultList.add(list.stream().skip(i*max).limit(max).collect(Collectors.toList()));
        });
        return resultList;
    }

    private static Integer countStep(int size, int max){
        return (size + max - 1) / max;
    }

}